package com.abyss.kafka;


import com.abyss.avro.AvroSerializerSchema;
import org.apache.avro.specific.SpecificRecordBase;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * @Date 2020/9/12
 */
public class KafkaPro<T extends SpecificRecordBase> {
    KafkaProducer producer = new KafkaProducer(getProp());
    /**
     * 1.创建类，泛型参数继承avro基类
     * 2.设置生产者参数
     * 3.自定avro序列化
     * 4.添加发送数据方法
     */

    //2.设置生产者参数
    public Properties getProp() {

        Properties properties = new Properties();
        /**
         * bootstrap.servers ：broker地址
         * acks ：0,1和-1
         * retries：重试次数
         * batch.size：批量发送大小默认16384 （16kB）
         * linger.ms： 定时发送1ms
         * buffer.memory: 缓存大小33554432
         * key.serializer：key序列化
         * value.serializer： value序列化
         */
        properties.put("bootstrap.servers", "node1:9092");
        properties.put("acks", "1");
        properties.put("retries", "0");
        properties.put("batch.size", "16384");
        properties.put("linger.ms", "1");
        properties.put("buffer.memory", "33554432");
        properties.put("key.serializer", StringSerializer.class.getName());
        properties.put("value.serializer", AvroSerializerSchema.class.getName());
        return properties;
    }

    // 4.添加发送数据方法
    public void send(String topic, T data){
        producer.send(new ProducerRecord(topic,data));
    }
}